package org.example.utils;


import cn.hutool.core.lang.Assert;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultPromise;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.Promise;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.reflect.FieldUtils;
import org.example.annotation.RpcReference;
import org.example.annotation.RpcServerScan;
import org.example.domain.po.RpcRequestMessagePO;
import org.example.handler.HeartBeatClientHandler;
import org.example.handler.RpcResponseMessageHandler;
import org.example.protocol.MessageCodec;
import org.example.protocol.ProcotolFrameDecoder;
import org.example.protocol.SequenceIdGenerator;
import org.example.registery.ServerRegistry;
import org.reflections.Reflections;

import java.lang.reflect.Field;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @description: 客户端负责发送请求，接收响应
 * @author: feixiang.li
 * @date: 2025-01-02 22:42
 */
@Slf4j
@Data
public class NettyClientUtils {

    /**
     * 根据序号key来判断是哪个请求的消息      value是用来接收结果的 promise 对象
     */
    private static final Map<Integer, Promise<Object>> PROMISES = new ConcurrentHashMap<Integer, Promise<Object>>();

    /**
     * channel集合  可能请求多个服务
     */
    private static final Map<String, Channel> channels = new ConcurrentHashMap<>();
    /**
     * 单例channel
     */
    private static final Bootstrap bootstrap = new Bootstrap();

    private static NioEventLoopGroup boss = new NioEventLoopGroup();

    private static ServerRegistry serverRegistry;

    private static Class<?> mainClass;


    /**
     * 获取promises
     *
     * @return promises
     */
    public static Map<Integer, Promise<Object>> getPromises() {
        return PROMISES;
    }

    public static void start(ServerRegistry registry, Class<?> clazz) {
        serverRegistry = registry;
        mainClass = clazz;
        startServer();
        createProxyService();
    }

    /**
     * 使用jdk动态代理创建代理对象
     */
    public static void createProxyService() {
        // 获取启动类main函数的路径
        String mainClassPath = mainClass.getName();
        // 获取main方法上 注解的值
        String annotationValue = mainClass.getAnnotation(RpcServerScan.class).value();
        //如果注解路径的值是空，则等于main父路径包下
        if ("".equals(annotationValue)) {
            annotationValue = mainClassPath.substring(0, mainClassPath.lastIndexOf("."));
        }
        //获取所有类的set集合
        Set<Class<?>> set = PackageScanUtils.getClasses(annotationValue);
        log.info("获取到类:{}个", set.size());
        for (Class<?> c : set) {
            try {
                scanField(c, annotationValue);
            } catch (Exception e) {
                log.error("扫描到RpcReference注解的类:{},异常信息", c, e);
            }
        }
    }

    public static void scanField(Class<?> clazz, String packageName) throws IllegalAccessException {
        Field[] allFields = FieldUtils.getAllFields(clazz);
        for (Field field : allFields) {
            // 拿到属性上的注解 RpcReference
            RpcReference rpcReference = field.getAnnotation(RpcReference.class);
            if (Objects.nonNull(rpcReference)) {
                // 拿到属性的类-接口
                Class<?> serviceClass = field.getType();
                log.error("扫描到RpcReference注解的类:{}", serviceClass);
                if (!serviceClass.isInterface()) {
                    return;
                }
                // 获取接口的实现类信息
                Class<?> subClassOne = getSubClassOne(serviceClass, packageName);
                ClassLoader loader = subClassOne.getClassLoader();
                Class<?>[] interfaces = subClassOne.getInterfaces();
                //创建代理对象
                Object o = Proxy.newProxyInstance(loader, interfaces, (proxy, method, args) -> {
                    // 1. 将方法调用转换为 消息对象
                    int sequenceId = SequenceIdGenerator.nextId();
                    RpcRequestMessagePO msg = new RpcRequestMessagePO(
                            sequenceId,
                            serviceClass.getName(),
                            method.getName(),
                            method.getReturnType(),
                            method.getParameterTypes(),
                            args
                    );
                    // 2. 准备一个空 Promise 对象，来接收结果 存入集合            指定 promise 对象异步接收结果线程
                    DefaultPromise<Object> promise = new DefaultPromise<Object>(boss.next());
                    NettyClientUtils.getPromises().put(sequenceId, promise);
                    // 3. 将消息对象发送出去
                    sendRpcRequest(msg);
                    // 4. 等待 promise 结果
                    promise.await();
                    if (promise.isSuccess()) {
                        // 调用正常
                        return promise.getNow();
                    } else {
                        // 调用失败
                        throw new RuntimeException(promise.cause());
                    }
                });
                // 把代理对象赋值到静态变量的属性中
                field.setAccessible(true);
                field.set(null, o);
            }
        }
    }

    public static <T> Class<? extends T>[] getSubClass(Class<T> clazz, String packageName) {
        Reflections reflections = new Reflections(packageName);
        Set<Class<? extends T>> subTypesOf = reflections.getSubTypesOf(clazz);
        return subTypesOf.toArray(new Class[subTypesOf.size()]);
    }

    public static <T> Class<? extends T> getSubClassOne(Class<T> clazz, String packageName) {
        Reflections reflections = new Reflections(packageName);
        Set<Class<? extends T>> subTypesOf = reflections.getSubTypesOf(clazz);
        return subTypesOf.iterator().next();
    }

    public static void sendRpcRequest(RpcRequestMessagePO msg) {
        InetSocketAddress service = serverRegistry.getService(msg.getInterfaceName());
        Channel channel = getChannel(service);
        // 判断channel是否可用
        if (!channel.isActive() || !channel.isRegistered()) {
            boss.shutdownGracefully();
            log.error("channel不可用-不能发送消息");
            return;
        }
        channel.writeAndFlush(msg).addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                log.debug("客户端发送消息成功");
            }
        });
    }

    public static void startServer() {
        bootstrap.group(boss)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new IdleStateHandler(0, 15, 0, TimeUnit.SECONDS));
                        //定长解码器
                        ch.pipeline().addLast(new ProcotolFrameDecoder());
                        // 添加消息解码器
                        ch.pipeline().addLast(new MessageCodec());
                        // 添加心跳处理器
                        ch.pipeline().addLast(new HeartBeatClientHandler());
                        // 添加rpc请求处理器
                        ch.pipeline().addLast(new RpcResponseMessageHandler());
                    }
                });
    }


    /**
     * 获取channel
     *
     * @param inetSocketAddress ip + port
     * @return channel
     */
    public static Channel getChannel(InetSocketAddress inetSocketAddress) {
        Assert.notNull(inetSocketAddress, "ip + port 不能为空");
        String key = inetSocketAddress.toString();
        // 判断是否已经连接
        if (channels.containsKey(key)) {
            Channel channel = channels.get(key);
            if (channel != null && channel.isActive()) {
                return channel;
            } else {
                channels.remove(key);
            }
        }
        Channel channel = null;
        try {
            channel = bootstrap.connect(inetSocketAddress).sync().channel();
            channel.closeFuture().addListener(new GenericFutureListener<Future<? super Void>>() {
                @Override
                public void operationComplete(Future<? super Void> future) throws Exception {
                    log.debug("断开连接");
                }
            });
            channels.put(key, channel);
            return channel;
        } catch (InterruptedException e) {
            if (channel != null) {
                channel.close();
            }
            Thread.currentThread().interrupt();
            log.error("连接失败", e);
        } catch (Exception e) {
            log.error("连接失败", e);
        }
        return null;
    }

}